package cn.doitedu.rtdw.data_etl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.util.EntityUtils;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;

public class Etl05_SearchActionAnalyse {
    public static void main(String[] args) throws Exception {
        // 编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 创建kafka连接器表，来映射kafka中的dwd层的用户行为公共维度宽表
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka (                           "
                        + "     user_id           BIGINT,                     "
                        + "     event_id          string,                     "
                        + "     event_time        bigint,                     "
                        + "     properties        map<string,string>,         "
                        + "     pt AS proctime() ,                            "
                        + "     rt AS to_timestamp_ltz(event_time,3),         "
                        + "     WATERMARK FOR rt AS rt - interval '0' second  "
                        + " ) WITH (                                          "
                        + "  'connector' = 'kafka',                           "
                        + "  'topic' = 'dwd-events-detail',                   "
                        + "  'properties.bootstrap.servers' = 'doitedu:9092', "
                        + "  'properties.group.id' = 'testGroup',             "
                        + "  'scan.startup.mode' = 'latest-offset',           "
                        + "  'value.format'='json',                           "
                        + "  'value.json.fail-on-missing-field'='false',      "
                        + "  'value.fields-include' = 'EXCEPT_KEY')           ");

        // 创建doris连接器表，映射doris中的搜索行为分析轻度聚合主题表
        tenv.executeSql(
                " create table search_doris (             "
                        + "     user_id          BIGINT,             "
                        + "     search_id        STRING,            "
                        + "     keyword          STRING,            "
                        + "     split_words      STRING,            "
                        + "     similar_word     STRING,            "
                        + "     search_time      BIGINT,            "
                        + "     return_item_count     BIGINT,       "
                        + "     click_item_count      BIGINT        "
                        + " ) WITH (                               "
                        + "    'connector' = 'doris',              "
                        + "    'fenodes' = 'doitedu:8030',         "
                        + "    'table.identifier' = 'dws.search_ana_agg',  "
                        + "    'username' = 'root',                "
                        + "    'password' = '',                    "
                        + "    'sink.label-prefix' = 'doris_tl" + System.currentTimeMillis() + "')"
        );


        /**
         * 按照5分钟滚动时间窗口，进行搜索行为聚合
         */
        tenv.executeSql(
                "  CREATE TEMPORARY VIEW agg_view AS                                         "
                        + "  WITH tmp AS (                                                   "
                        + "  SELECT                                                          "
                        + "    user_id,                                                      "
                        + "    event_time,                                                   "
                        + "    event_id,                                                     "
                        + "    properties['search_id'] as search_id,                         "
                        + "    properties['keyword'] as keyword,                             "
                        + "    cast(properties['res_cnt'] as INT) as res_cnt,                "
                        + "    properties['item_seq'] as item_seq,                           "
                        + "    rt                                                            "
                        + "  from dwd_kafka                                                  "
                        + "  WHERE event_id in ('search','search_click','search_return')     "
                        + "  )                                                               "
                        + "                                                                  "
                        + "  SELECT                                                          "
                        + "     user_id,                                                     "
                        + "     search_id,                                                   "
                        + "     keyword,                                                     "
                        + "     '' as split_words,                                           "
                        + "     '' as similar_word,                                          "
                        + "     min(event_time) as search_time,                              "
                        + "     max(res_cnt) as return_item_count,                           "
                        + "     count(item_seq) as click_item_count                          "
                        + "  FROM TABLE(                                                     "
                        + "    TUMBLE(TABLE tmp,DESCRIPTOR(rt),INTERVAL '5' MINUTE)          "
                        + "  )                                                               "
                        + "  GROUP BY                                                        "
                        + "     window_start,                                                "
                        + "     window_end,                                                  "
                        + "     user_id,                                                     "
                        + "     search_id,                                                   "
                        + "     keyword                                                      "
        );


        /**
         * 对聚合好的数据中的搜索词，去外部http接口请求它的分词结果和同义词结果
         */
        DataStream<Etl05Bean> aggStream = tenv.toDataStream(tenv.from("agg_view"), Etl05Bean.class);
        //aggStream.print();

        SingleOutputStreamOperator<Etl05Bean> resultStream = aggStream.keyBy(b -> b.getKeyword())
                .process(new KeyedProcessFunction<String, Etl05Bean, Etl05Bean>() {

                    CloseableHttpClient client;
                    HttpPost post;

                    MapState<String, String> state;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        client = HttpClientBuilder.create().build();

                        // 构造请求参数对象
                        post = new HttpPost("http://doitedu:8081/api/post/simwords");
                        post.addHeader("Content-type", "application/json; charset=utf-8");
                        post.addHeader("Accept", "application/json");


                        // 声明状态来存储已经请求过的近义词结果
                        // 搜索词 -> 近义词\001分词
                        state = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("st", String.class, String.class));

                    }

                    @Override
                    public void processElement(Etl05Bean bean, KeyedProcessFunction<String, Etl05Bean, Etl05Bean>.Context ctx, Collector<Etl05Bean> out) throws Exception {
                        // 取出本条数据中的keyword
                        String keyword = bean.getKeyword();
                        String stateResult = state.get(keyword);

                        if (stateResult == null) {
                            // 根据keyword，去请求公司内的http接口
                            HashMap<String, String> data = new HashMap<>();
                            data.put("origin", keyword);

                            post.setEntity(new StringEntity(JSON.toJSONString(data), StandardCharsets.UTF_8));

                            // 用客户端对象执行请求
                            CloseableHttpResponse response = client.execute(post);
                            HttpEntity entity = response.getEntity();
                            String resultJson = EntityUtils.toString(entity);
                            JSONObject resultJsonObj = JSON.parseObject(resultJson);

                            // 从响应结果json中，取出我们需要的分词和近义词
                            String split_words = resultJsonObj.getString("words");
                            String similar_word = resultJsonObj.getString("similarWord");

                            // 将接口请求到的结果，缓存到状态中
                            state.put(keyword, split_words + "\001" + similar_word);

                            // 将分词和近义词填充到数据bean中，并输出
                            bean.setSplit_words(split_words);
                            bean.setSimilar_word(similar_word);
                        } else {

                            String[] split = stateResult.split("\001");
                            bean.setSplit_words(split[0]);
                            bean.setSimilar_word(split[1]);
                        }


                        out.collect(bean);

                    }
                });


        /**
         * 将上述得到的完整结果流，转成表
         */
        tenv.createTemporaryView("res",resultStream);


        /**
         * 从结果视图中select数据，然后insert到doris连接器表
         */
        tenv.executeSql("INSERT INTO search_doris select * from res ");


        env.execute();
    }
}
